package com.hitqz.robot.biz.websocket.endpoint;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.collection.ConcurrentHashSet;
import cn.hutool.core.util.HashUtil;
import jakarta.websocket.*;
import jakarta.websocket.server.ServerEndpoint;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArraySet;

@ServerEndpoint(value = "/websocket/live")
@Component
@Slf4j
public class LiveEndpoint {
	// 存储sourceKey和会话的对应关系
	private static final Map<String, Set<Session>> sourceKeySessionMap = new ConcurrentHashMap<>();
	// 用于存储所有打开的WebSocket sessions
	private static final Set<Session> sessions = new CopyOnWriteArraySet<>();
	// 用于存储视频源地址
	private static final Set<String> sourcesKeySet = new ConcurrentHashSet<>();


	@OnOpen
	public void onOpen(Session session) {
		sessions.add(session);
		log.debug("[ws][live]session-id:{} connected.current size :{}",session.getId(),sessions.size());
		String queryString = session.getQueryString();
		String[] pairs = queryString.split("&");
		Map<String, String> queryParams = new HashMap<>();
		for (String pair : pairs) {
			// 使用=分割键和值
			int idx = pair.indexOf("=");
			// 将键值对添加到Map中
			queryParams.put(pair.substring(0, idx), pair.substring(idx + 1));
		}
		String source = queryParams.get("source");
		String sourceKey = String.valueOf(HashUtil.javaDefaultHash(source));
		//保存 sourceKey和session的对应关系 一对多
		sourceKeySessionMap.computeIfAbsent(sourceKey, k -> ConcurrentHashMap.newKeySet()).add(session);
		if (sourcesKeySet.add(sourceKey)){
			// 开始推流
			log.info("[ws][live]sourceKey:{},source:{}，start push", sourceKey, source);
		}
	}

	@OnMessage
	public void onMessage(Session session,String message) {
		log.info("[ws][live]receive session-id:{},message:{}",session.getId(),message);
	}

	@OnClose
	public void onClose(Session session) {
		// 移除session
		sessions.remove(session);
		log.info("[ws][live]session-id:{},disconnected.current size :{}",session.getId(),sessions.size());
		for (Set<Session> sessions : sourceKeySessionMap.values()) {
			sessions.remove(session);
		}
	}

	@OnError
	public void onError(Session session,Throwable error) {
		log.error("[ws][live]session-id:{},error:{}", session.getId(),error.getMessage());
	}

	// 发送二进制数据
	public static void sendVideoData(String sourceKey, ByteBuffer data){
		for (Session session : CollUtil.emptyIfNull(sourceKeySessionMap.get(sourceKey))) {
			if (session.isOpen()) {
				session.getAsyncRemote().sendBinary(data);
			}
		}
	}


}
